# -*- coding: utf-8 -*-

# 导入必要的库
from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType


# 定义函数
def spark_analyse(filename):

    print("开始spark分析")

    # 创建SparkSession
    spark = SparkSession.builder.\
        master("local").\
        appName("rent_als"). \
        config("spark.sql.shuffle.partitions", 2). \
        getOrCreate()

    # 读取CSV文件
    df = spark.read.csv(filename, header=True)

    # 初始化列表
    # 0雁塔、1碑林、2莲湖、3未央、4新城、5长安、6灞桥、7高新、8曲江新区、9浐灞、10经开、11航天新城、12西咸新区
    max_list = [0 for i in range(13)]  # max_list存储各个区的最大值
    mean_list = [1.2 for i in range(13)]  # 用于存储各个区的平均值
    min_list = [0 for i in range(13)]  # 用于存储各个区的最小值
    mid_list = [0 for i in range(13)]  # 用于存储各个区的中位数

    # 类型转换，十分重要，保证了price列作为int用来比较，否则会用str比较, 同时排除掉一些奇怪的价格，比如写字楼的出租超级贵
    # 或者有人故意标签1元，其实要面议, 还有排除价格标记为面议的
    df = df.filter(df.price != '面议').withColumn("price", df.price.cast(IntegerType()))
    df = df.filter(df.price >= 50).filter(df.price <= 40000)

    # 计算各个区的最大值
    max_list[0] = df.filter(df.area == "雁塔").agg({"price": "max"}).first()['max(price)']
    max_list[1] = df.filter(df.area == "碑林").agg({"price": "max"}).first()['max(price)']
    max_list[2] = df.filter(df.area == "莲湖").agg({"price": "max"}).first()['max(price)']
    max_list[3] = df.filter(df.area == "未央").agg({"price": "max"}).first()['max(price)']
    max_list[4] = df.filter(df.area == "新城").agg({"price": "max"}).first()['max(price)']
    max_list[5] = df.filter(df.area == "长安").agg({"price": "max"}).first()['max(price)']
    max_list[6] = df.filter(df.area == "灞桥").agg({"price": "max"}).first()['max(price)']
    max_list[7] = df.filter(df.area == "高新").agg({"price": "max"}).first()['max(price)']
    max_list[8] = df.filter(df.area == "曲江新区").agg({"price": "max"}).first()['max(price)']
    max_list[9] = df.filter(df.area == "浐灞"    ).agg({"price": "max"}).first()['max(price)']
    max_list[10] = df.filter(df.area == "经开"   ).agg({"price": "max"}).first()['max(price)']
    max_list[11] = df.filter(df.area == "航天新城").agg({"price": "max"}).first()['max(price)']
    max_list[12] = df.filter(df.area == "西咸新区").agg({"price": "max"}).first()['max(price)']

    # 计算各个区的最小值
    min_list[0] = df.filter(df.area == "雁塔").agg({"price": "min"}).first()['min(price)']
    min_list[1] = df.filter(df.area == "碑林").agg({"price": "min"}).first()['min(price)']
    min_list[2] = df.filter(df.area == "莲湖").agg({"price": "min"}).first()['min(price)']
    min_list[3] = df.filter(df.area == "未央").agg({"price": "min"}).first()['min(price)']
    min_list[4] = df.filter(df.area == "新城").agg({"price": "min"}).first()['min(price)']
    min_list[5] = df.filter(df.area == "长安").agg({"price": "min"}).first()['min(price)']
    min_list[6] = df.filter(df.area == "灞桥").agg({"price": "min"}).first()['min(price)']
    min_list[7] = df.filter(df.area == "高新").agg({"price": "min"}).first()['min(price)']
    min_list[8] = df.filter(df.area == "曲江新区" ).agg({"price": "min"}).first()['min(price)']
    min_list[9] = df.filter(df.area == "浐灞"    ).agg({"price": "min"}).first()['min(price)']
    min_list[10] = df.filter(df.area == "经开"   ).agg({"price": "min"}).first()['min(price)']
    min_list[11] = df.filter(df.area == "航天新城").agg({"price": "min"}).first()['min(price)']
    min_list[12] = df.filter(df.area == "西咸新区").agg({"price": "min"}).first()['min(price)']

    # 计算各个区的平均值
    mean_list[0] = int(df.filter(df.area == "雁塔").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[1] = int(df.filter(df.area == "碑林").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[2] = int(df.filter(df.area == "莲湖").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[3] = int(df.filter(df.area == "未央").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[4] = int(df.filter(df.area == "新城").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[5] = int(df.filter(df.area == "长安").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[6] = int(df.filter(df.area == "灞桥").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[7] = int(df.filter(df.area == "高新").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[8] = int(df.filter(df.area == "曲江新区" ).agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[9] = int(df.filter(df.area == "浐灞"    ).agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[10] = int(df.filter(df.area == "经开"   ).agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[11] = int(df.filter(df.area == "航天新城").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)
    mean_list[12] = int(df.filter(df.area == "西咸新区").agg({"price": "mean"}).collect()[0]['avg(price)'] // 1)

    # 计算各个区的中位数
    mid_list[0] = df.filter(df.area == "雁塔").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[1] = df.filter(df.area == "碑林").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[2] = df.filter(df.area == "莲湖").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[3] = df.filter(df.area == "未央").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[4] = df.filter(df.area == "新城").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[5] = df.filter(df.area == "长安").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[6] = df.filter(df.area == "灞桥").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[7] = df.filter(df.area == "高新").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[8] = df.filter(df.area == "曲江新区" ).approxQuantile("price", [0.5], 0.01)[0]
    mid_list[9] = df.filter(df.area == "浐灞"    ).approxQuantile("price", [0.5], 0.01)[0]
    mid_list[10] = df.filter(df.area == "经开"   ).approxQuantile("price", [0.5], 0.01)[0]
    mid_list[11] = df.filter(df.area == "航天新城").approxQuantile("price", [0.5], 0.01)[0]
    mid_list[12] = df.filter(df.area == "西咸新区").approxQuantile("price", [0.5], 0.01)[0]

    # 构建结果列表
    all_list = []
    all_list.append(max_list)
    all_list.append(min_list)
    all_list.append(mean_list)
    all_list.append(mid_list)

    print("结束spark分析")

    return all_list